package com.ysx.article.stream;

import com.alibaba.fastjson.JSON;
import com.ysx.article.config.KafkaStreamListener;
import com.ysx.article.dto.ArticleVisitStreamMess;
import com.ysx.behaviour.dto.UpdateArticleMess;
import com.ysx.common.constants.BusinessConstants;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Arrays;

@Component
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String, String>> {

    //接收来自生产者发送的消息的主题
    @Override
    public String listenerTopic() {
        return BusinessConstants.MqConstants.HOT_ARTICLE_SCORE_TOPIC;
    }

    //聚合了结果之后进行发送出去的主题
    @Override
    public String sendTopic() {
        return BusinessConstants.MqConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC;
    }

    //处理的业务
    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {
        //核心的逻辑 就是获取到当前的发送过来的哪一篇文章的 点赞数 和 评论数 等
        //先获取到value的值

        //将其转换成POJO
        //获取到里面的文章的ID 和操作类型 进行聚合 要聚合 就先构建key : articleId:type--->1
        //再进行发送，再发送之前先进行设置发送出去的数据 再发送
        //textLine -> Arrays.asList(textLine.toLowerCase().split(","))

        KTable<Windowed<String>, Long> wordCounts = stream
                .flatMapValues(value -> {
                    //获取到JSON的数据
                    UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                    //进行聚合 按照 articleId:type 进行统计
                    String s = mess.getArticleId() + ":" + mess.getType().name();
                    return Arrays.asList(s);
                })
                //设置根据word来进行统计 而不是根据key来进行分组
                .groupBy(new KeyValueMapper<String, String, String>() {
                    @Override
                    public String apply(String key, String value) {
                        System.out.println("我看看聚合是靠什么:" + value);
                        return value;
                    }
                })
                //设置5秒窗口时间
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5L)))
                //进行count统计
                .count(Materialized.as("counts-store"));
        //将统计后的数据再次发送到消息主题中
        //变成流 发送给  发送的状态设置为 将数据转成字符串？为什么呢。因为我们的数据kafka接收都是字符串了
        return wordCounts
                .toStream()
                .map((key, value) -> {
                    //value 是数值
                    //key: 123:LIKES
                    System.out.println(key.key().toString() + ":::::" + value);

                    //注意 需要发送到输出的topic的时候需要进行设置 进行封装
                    String str = key.key().toString();

                    String[] split = str.split(":");
                    ArticleVisitStreamMess articleVisitStreamMess = new ArticleVisitStreamMess();
                    articleVisitStreamMess.setArticleId(Long.valueOf(split[0]));
                    switch (UpdateArticleMess.UpdateArticleType.valueOf(split[1])) {

                        case LIKES: {
                            articleVisitStreamMess.setLike(Long.valueOf(value));
                            break;
                        }
                        case COLLECTION: {
                            articleVisitStreamMess.setCollect(Long.valueOf(value));
                            break;
                        }
                        case COMMENT: {
                            articleVisitStreamMess.setComment(Long.valueOf(value));
                            break;
                        }

                        case VIEWS: {
                            articleVisitStreamMess.setView(Long.valueOf(value));
                            break;
                        }
                        default: {
                            System.out.println("啥也没有");
                            break;
                        }
                    }

                    //发送出去  消息本身的内容就是一个JSON的字符串
                    return new KeyValue(key.key().toString(), JSON.toJSONString(articleVisitStreamMess));
                });
    }
}